package org.example

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.desc
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}

object ketang6 {
    def main(args: Array[String]): Unit = {
      val spark = SparkSession.builder().master("local[*]").appName("spark").getOrCreate()
      val sc = spark.sparkContext
      //表结构读取电影用户数据
      val schemaUser = StructType(Seq(
        StructField("id", IntegerType, nullable = false),
        StructField("gender", StringType),
        StructField("age", StringType),
        StructField("occupation", IntegerType),
        StructField("location", StringType)
      ))
      val user =spark.read.option("sep","::").schema(schemaUser)
        .csv("src/main/resources/users.dat")
      user.show(5)
      /*//1、查询 替换  where/filter
      user.where("gender ='F' and age=18").show(3)
      user.filter("gender ='F' and age=18").show(3)

      spark.udf.register("replace",(x:String) =>{
        x match {
          case "M" => 0
          case "F" => 1
        }
      })
      user.selectExpr("id","replace(gender) as sexual","age").show(3)
      user.select("id","age","location").show(3)
      //2、排序
      user.orderBy(desc("age")).show(5)
      user.sort(-user("id")).show(5)
      //3、分组
      user.groupBy("gender").count().show()*/
      //4、连接 join(DataFrame,“列名”，“连接方式left_outer”
      //5、练习：求18岁女生评分电影为5分的所有电影名
      // 读取评分数据
      val schemaRating = StructType(Seq(
        StructField("userId", IntegerType, nullable = false),
        StructField("movieId", IntegerType, nullable = false),
        StructField("rating", IntegerType),
        StructField("timestamp", StringType)
      ))
      val rating = spark.read.option("sep", "::").schema(schemaRating)
        .csv("src/main/resources/ratings.dat")

      // 读取电影数据
      val schemaMovie = StructType(Seq(
        StructField("movieId", IntegerType, nullable = false),
        StructField("title", StringType),
        StructField("genres", StringType)
      ))
      val movie = spark.read.option("sep", "::").schema(schemaMovie)
        .csv("src/main/resources/movies.dat")

      // 查询18岁女生评分电影为5分的所有电影名
      val result = user
        .where("gender = 'F' and age = '18'")
        .join(rating, user("id") === rating("userId"))
        .where("rating = 5")
        .join(movie, rating("movieId") === movie("movieId"))
        .select("title")
        .distinct()

      result.show()

      sc.stop()
    }
}
